package pro.coman404.tools;

import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;

import java.io.IOException;

/**
 * @author NotFound404 at 2024/7/1
 * code code always code
 */
public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {

    private ValueState<Boolean> flagState;
    private ValueState<Long> timerState;

    private static final double SMALL_AMOUNT = 1D;
    private static final double LARGE_AMOUNT = 500D;

    private static final long ONE_MINUTE = 60 * 1000;

    @Override
    public void open(OpenContext openContext) throws Exception {

        ValueStateDescriptor<Boolean> flagStateDescriptor = new ValueStateDescriptor<>("flag", Types.BOOLEAN);
        flagState = getRuntimeContext().getState(flagStateDescriptor);

        ValueStateDescriptor<Long> timerStateDescriptor = new ValueStateDescriptor<>("timer-state", Types.LONG);
        timerState = getRuntimeContext().getState(timerStateDescriptor);
    }

    @Override
    public void processElement(Transaction value, KeyedProcessFunction<Long, Transaction, Alert>.Context ctx, Collector<Alert> out) throws Exception {
        double amount = value.getAmount();
        Boolean lastTransactionWasSmall = flagState.value();
        if(lastTransactionWasSmall!=null){
            if(amount > LARGE_AMOUNT){
                Alert alert = new Alert();
                alert.setId(value.getAccountId());
                out.collect(alert);
            }
//            flagState.clear();
            this.cleanup(ctx);
        }

        if (amount < SMALL_AMOUNT){
            flagState.update(true);

            //开始计时
            long timer = ctx.timerService().currentProcessingTime() + ONE_MINUTE;
            ctx.timerService().registerProcessingTimeTimer(timer);
            timerState.update(timer);
        }

    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<Long, Transaction, Alert>.OnTimerContext ctx, Collector<Alert> out) throws Exception {
        timerState.clear();
        flagState.clear();
    }

    private void cleanup(Context context) throws IOException {
        Long timer = timerState.value();
        if (timer!=null){
            context.timerService().deleteEventTimeTimer(timer);
        }
        timerState.clear();
        flagState.clear();
    }
}
